package com.chis.flink;


import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;


public class ScoketCount {

    public static void main(String[] args) {
        test2();
    }

    public static void test2(){
        try {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            DataStreamSource<String> dataStreamSource = env.socketTextStream("127.0.0.1",9000,"\n");

            DataStream dataStream = dataStreamSource.map(new MapFunction<String, String>() {
                @Override
                public String map(String val){
                    return val.toUpperCase();
                }
            });
            dataStream.addSink(new SinkFunction<String>() {
                @Override
                public void invoke(String value) throws Exception {
                    System.out.println(value);
                    
                }
            });

            env.execute("test");
        } catch (Exception e){
            e.printStackTrace();
        }
    }

    public static void test1(){
        try {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            DataStreamSource<String> dataStreamSource = env.socketTextStream("127.0.0.1",9000,"\n");
            DataStream<Tuple2<String,Integer>> wordcount =  dataStreamSource.flatMap((String s, Collector<Tuple2<String,Integer>> collector) -> {
                for (String word : s.split("\\s")) {
                    collector.collect(Tuple2.of(word,1));
                }
            }).returns(Types.TUPLE(Types.STRING,Types.INT));
            //流式处理
            DataStream<Tuple2<String, Integer>> windowCounts = wordcount.keyBy(0).sum(1);
            //10秒翻滚窗，每次只统计最新10秒内的数据
//            DataStream<Tuple2<String, Integer>> windowCounts = wordcount.keyBy(0).timeWindow(Time.seconds(10)).sum(1);
            //5秒的滑动窗口，每隔5秒统计过去10秒的数据
//            DataStream<Tuple2<String, Integer>> windowCounts = wordcount.keyBy(0).timeWindow(Time.seconds(5), Time.seconds(2)).sum(1);

            windowCounts.print().setParallelism(1);
            env.execute("socket window count");
        } catch (Exception e){
            e.printStackTrace();
        }
    }


}
